package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.VisitorStats;
import com.atguigu.realtime.common.ConstantTopic;
import com.atguigu.realtime.util.MyKafkaUtil;
import com.atguigu.realtime.util.MySinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Map;

/**
 * @ClassName: DWSVisitorStatsApp
 * @Description:
 * @Author: kele
 * @Date: 2021/4/25 10:22
 * <p>
 * <p>
 * <p>
 * 得到用户宽表
 **/
public class DWSVisitorStatsApp extends BaseAppV2 {


    public static void main(String[] args) throws SQLException, ClassNotFoundException {

        new DWSVisitorStatsApp().init(40001,
                2,
                "DWSVisitorStatsApp",
                "DWSVisitorStatsApp",
                ConstantTopic.DWD_PAGE,
                ConstantTopic.DWM_UV,
                ConstantTopic.DWM_USER_JUMP_DETAIL);

    }

    @Override
    protected void run(StreamExecutionEnvironment env,
                       Map<String, DataStream<String>> streams) throws SQLException, ClassNotFoundException {

        /**
         * 获取三个流，将其封装为visitorStats，将其合并为一个流
         */
        DataStream<VisitorStats> visitorStatsDataStream = merge3SteamsToVisitorStream(streams);

        //visitorStatsDataStream.print();

        /**
         * 对合并的流按照维度进行聚合，
         * 按照 窗口 和 维度 进行聚合（维度选多一些，可以再按照维度进行聚合）
         */
        SingleOutputStreamOperator<VisitorStats> aggStream = aggbyWindowandDim(visitorStatsDataStream);


        /**
         * 将数据写入到clickhouse中
         */

        send2Clickhouse(aggStream);

    }

    private void send2Clickhouse(SingleOutputStreamOperator<VisitorStats> aggStream) {

        aggStream.addSink(MySinkUtil
                .getClickHouseSink("gmall2021","visitor_stats_2021",VisitorStats.class));
    }

    /**
     * 将得到的一个流对窗口和维度进行聚合
     *
     * @param visitorStatsDataStream
     */
    private SingleOutputStreamOperator<VisitorStats> aggbyWindowandDim(DataStream<VisitorStats> visitorStatsDataStream) {

        return visitorStatsDataStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTs())
                )
                //按照所有的维度进行聚合
                .keyBy(v -> v.getVc() + "_" + v.getCh() + "_" + v.getAr() + "_" + v.getIs_new())
                //得到5s的窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<VisitorStats>() {

                            @Override
                            public VisitorStats reduce(VisitorStats value1, VisitorStats value2) throws Exception {

                                value1.setUv_ct(value1.getUv_ct() + value2.getUv_ct());
                                value1.setPv_ct(value1.getPv_ct() + value2.getPv_ct());
                                value1.setSv_ct(value1.getSv_ct() + value2.getSv_ct());
                                value1.setUj_ct(value1.getUj_ct() + value2.getUj_ct());
                                value1.setDur_sum(value1.getDur_sum() + value2.getDur_sum());
                                return value1;
                            }
                        },

                        //在reduce方法之后，可以重写方法，添加字段
                        //添加字段：stt，edt段，修改ts字段
                        new ProcessWindowFunction<VisitorStats, VisitorStats, String, TimeWindow>() {

                            private SimpleDateFormat sdf;

                            @Override
                            public void open(Configuration parameters) throws Exception {

                                sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                            }

                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<VisitorStats> elements, //聚合的结果，只有一条
                                                Collector<VisitorStats> out) throws Exception {


                                VisitorStats visitorStats = elements.iterator().next();

                                long start = ctx.window().getStart();
                                long end = ctx.window().getEnd();

                                visitorStats.setStt(sdf.format(start));
                                visitorStats.setEdt(sdf.format(end));
                                visitorStats.setTs(end);

                                out.collect(visitorStats);
                            }
                        }
                );
    }

    private DataStream<VisitorStats> merge3SteamsToVisitorStream(Map<String, DataStream<String>> streams) {

        DataStream<String> pageStream = streams.get(ConstantTopic.DWD_PAGE);
        DataStream<String> uvStream = streams.get(ConstantTopic.DWM_UV);
        DataStream<String> jumpStream = streams.get(ConstantTopic.DWM_USER_JUMP_DETAIL);

        //通过page流：得到  页面访问数，持续访问时间
        SingleOutputStreamOperator<VisitorStats> TransfromPageStream = pageStream
                .map(value -> {

                    JSONObject json = JSON.parseObject(value);
                    JSONObject common = json.getJSONObject("common");
                    JSONObject page = json.getJSONObject("page");

                    return new VisitorStats(
                            "", "",
                            common.getString("vc"),
                            common.getString("ch"),
                            common.getString("ar"),
                            common.getString("is_new"),
                            //独立访客数，页面访问数，进入次数，跳出次数，持续时间，统计时间
                            0L, 1L, 0L, 0L,
                            page.getLong("during_time"),
                            json.getLong("ts")
                    );
                });

        //通过uv流，得到  独立访客数
        SingleOutputStreamOperator<VisitorStats> TransfromUVStream =
                uvStream
                .map(value -> {

                    JSONObject json = JSON.parseObject(value);
                    JSONObject common = json.getJSONObject("common");

                    return new VisitorStats(
                            "", "",
                            common.getString("vc"),
                            common.getString("ch"),
                            common.getString("ar"),
                            common.getString("is_new"),
                            //独立访客数，页面访问数，进入次数，跳出次数，持续时间，访问时间
                            1L, 0L, 0L, 0L,
                            0L,
                            json.getLong("ts")

                    );
                });

        //通过过滤的页面日志，得到 进入次数
        SingleOutputStreamOperator<VisitorStats> TransfromSVStream =
                pageStream
                        .flatMap(new FlatMapFunction<String, VisitorStats>() {
                            @Override
                            public void flatMap(String value, Collector<VisitorStats> out) throws Exception {
                                JSONObject json = JSON.parseObject(value);
                                JSONObject common = json.getJSONObject("common");

                                String last_page = json.getJSONObject("page").getString("last_page_id");

                                //当没有前一页时，则属于进入
                                if (last_page == null || last_page.isEmpty()) {
                                    VisitorStats visitorStats = new VisitorStats(
                                            "", "",
                                            common.getString("vc"),
                                            common.getString("ch"),
                                            common.getString("ar"),
                                            common.getString("is_new"),
                                            //独立访客数，页面访问数，进入次数，跳出次数，持续时间，访问时间
                                            0L, 0L, 1L, 0L,
                                            0L,
                                            json.getLong("ts"));

                                    out.collect(visitorStats);
                                }
                            }
                        });


        //通过跳出流得到跳出次数
        SingleOutputStreamOperator<VisitorStats> TransfromJumpStream =
                jumpStream
                        .map(value -> {
                            JSONObject json = JSON.parseObject(value);
                            JSONObject common = json.getJSONObject("common");

                            return new VisitorStats(
                                    "", "",
                                    common.getString("vc"),
                                    common.getString("ch"),
                                    common.getString("ar"),
                                    common.getString("is_new"),
                                    //独立访客数，页面访问数，进入次数，跳出次数，持续时间，访问时间
                                    0L, 0L, 0L, 1L,
                                    0L,
                                    json.getLong("ts")
                            );
                        });


        //将四个流合并输出
        DataStream<VisitorStats> union = TransfromPageStream.union(TransfromUVStream, TransfromSVStream, TransfromJumpStream);

        return union;

    }
}
